feat: add distributed map operation by nvasiu · Pull Request #1 · nvasiu/aws-durable-execution-sdk-python

nvasiu · 2026-08-20T21:10:33Z

Summary

Adds the distributed map operation (ctx.distributed_map) to the Python SDK:
A map run processes a bounded dataset in parallel. A customer starts a map run from a durable function, naming a source to read items from, a processor function to invoke per batch, and concurrency, retry, and failure settings. The service reads items from the source, groups them into batches, invokes the processor for each batch, retries failures, tracks progress, routes successful results and failed items to destinations, and reports completion.

Changes

concurrency/models.py

DistributedMapSummary: what ctx.distributed_map returns, describes run's overall outcome.
DistributedMapResult: returned from ctx.distributed_map when inline result collection is enabled. Contains individual map run item outcomes.
DistributedMapResultItem and DistributedMapItemError: represent a single item's result / error

config.py

DistributedMapConfig: optional settings for distributed map
DistributedMapSource: describes where map run items come from (inline list, S3, or a custom reader)
DistributedMapProcessor: describes the Lambda that processes items and how outcomes are reported back
ProcessorRetryConfig: configures how failing items are retried
DistributedMapCompletionConfig: defines item failure thresholds for marking the overall map run failed
SuccessDestination, FailureDestination, DistributedMapDestinationConfig, DistributedMapDestination: for routing successful and failed item records to S3

context.py

ctx.distributed_map: the entry point a customer calls to run a distributed map

distributed_map_helpers.py

Authoring wrappers: let a customer write a plain function and have it work as a processor Lambda without hand-writing the item or batch protocol, including durable-execution variants and a reader
Currently placed in a separate top level file, can be moved elsewhere.

operation/distributed_map.py

The executor: drives the operation so the caller's function suspends while the run executes and resumes with the finished outcome, and surfaces a clear error if the operation itself fails

lambda_service.py

Carries the operation and its results to and from the backend service

state.py

Stores the run's outcome in the durable execution state so it persists across suspend and resume

exceptions.py

DistributedMapError: the error a customer catches when a run or an item fails

__init__.py

Makes the distributed-map types importable by customers as public API

Tests

tests/operation/distributed_map_test.py
tests/context_test.py

Core operation unit tests: executor, config/argument validation, wire round trips, result types

tests/e2e/distributed_map_int_test.py

End to end ctx.distributed_map tests, mocking backend responses: suspend / resume, collect results, throw on failure

tests/distributed_map_helpers_test.py

Authoring wrapper tests: checking that they process items, report failures, reject bad inputs

tests/e2e/distributed_map_helpers_int_test.py

End to end authoring wrapper tests.

TODO

When the model changes and distributed map implementation are complete in the durable service backend, these SDK changes need to be verified against them.

Future Tasks

Add distributed map to the local emulator (in the testing package).
- When this is done, we can add full end to end tests using the emulator.
Add distributed map examples to the examples package.

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.

- Add ctx.distributed_map with inline, S3, and reader sources - Add DistributedMapConfig, processor, completion, and destination config types - Add DistributedMapResult/Summary result types and DistributedMapError - Add function-authoring helpers for item and batch handlers - Serialize the DISTRIBUTED_MAP operation and add its executor

yaythomas

this won't work against installed botocore yet. the checkpoint goes through the boto3 lambda client, which rejects unknown structure members at param-validation time, so every ctx.distributed_map call fails client-side with ParamValidationError until a botocore release ships the distributed-map model.

bump the boto3 minimum pin once this available. so the failure mode becomes an install-time constraint instead of a runtime one.

yaythomas · 2026-08-20T22:34:05Z

+        """Processor that reports the ids of failed items, with all others marked succeeded."""
+        return cls(
+            function_name=name,
+            response_mode="ReportBatchItemFailures",


this correct? vs REPORT_BATCH_ITEM_FAILURES ?

shouldn't this be captured in an enum?

enum DistributedMapFunctionResponseType { REPORT_BATCH_ITEM_FAILURES REPORT_BATCH_ITEM_RESULTS }

which would have helped prevent the drift between wire-value and serialization classes.

yaythomas · 2026-08-20T22:48:33Z

+class DistributedMapSource:
+    """Source configuration for a map run."""
+
+    source_type: str


suggestion: this would benefit from being an enum rather than a raw string, matching the other wire enums in this SDK (e.g. OperationType, OperationStatus).

yaythomas · 2026-08-20T22:48:34Z

+    bucket: str
+    key: str | None = None
+    prefix: str | None = None
+    transform: str | None = None


suggestion: this would benefit from being an enum rather than a raw string, matching the other wire enums in this SDK (e.g. OperationType, OperationStatus).

yaythomas · 2026-08-20T22:48:35Z

+    key: str | None = None
+    prefix: str | None = None
+    transform: str | None = None
+    fmt: str | None = None


suggestion: this would benefit from being an enum rather than a raw string, matching the other wire enums in this SDK (e.g. OperationType, OperationStatus).

yaythomas · 2026-08-20T22:48:36Z

+            uri: str,
+            *,
+            headers: Sequence[str] | None = None,
+            delimiter: str = "COMMA",


suggestion: this would benefit from being an enum rather than a raw string, matching the other wire enums in this SDK (e.g. OperationType, OperationStatus).

yaythomas · 2026-08-20T23:56:55Z

    WAIT = "WAIT"
    CALLBACK = "CALLBACK"
    CHAINED_INVOKE = "CHAINED_INVOKE"
+    DISTRIBUTED_MAP = "DISTRIBUTED_MAP"


any other changes necessary here to wire the events in for plugin? are there specific events?

yaythomas · 2026-08-21T00:02:54Z

@@ -0,0 +1,502 @@
+"""Implement the Durable map run operation."""


module name could be dmap, avoid underscore

yaythomas · 2026-08-21T00:03:20Z

+def _s3_config_to_wire(s3: S3SourceConfig) -> dict[str, Any]:
+    """Translate a resolved S3 source config into its wire dict."""
+    result: dict[str, Any] = {"Bucket": s3.bucket}
+    if s3.key is not None:


repo convention is for this to live on the model

yaythomas · 2026-08-21T00:06:17Z

@@ -0,0 +1,241 @@
+"""Authoring helpers for distributed map processor and reader Lambda functions.


how about dmap.py, mirroring the existing waits.py / operation/wait.py split? keeping it un-exported from the root package (as you already do) seems right.

yaythomas · 2026-08-21T00:08:54Z

 logger = logging.getLogger(__name__)


+@functools.lru_cache(maxsize=1)


is reading the pattern out of botocore's service model worth it here vs a simpler non-empty/length check?
the fine grammar gets enforced by the service anyway, and if botocore ever drops pattern/max from the shape metadata this starts raising KeyError on every processor construction.

nvasiu deployed to ai-pr-review August 20, 2026 21:10 — with GitHub Actions Active

nvasiu had a problem deploying to ai-pr-review-runtime August 20, 2026 21:10 — with GitHub Actions Failure

nvasiu force-pushed the feat/map-run branch from 4b93b1a to 535d35c Compare August 20, 2026 21:27

nvasiu deployed to ai-pr-review August 20, 2026 21:27 — with GitHub Actions Active

nvasiu had a problem deploying to ai-pr-review-runtime August 20, 2026 21:27 — with GitHub Actions Failure

yaythomas suggested changes Aug 21, 2026

View reviewed changes

Navigation Menu

feat: add distributed map operation - #1

feat: add distributed map operation#1
nvasiu wants to merge 1 commit into
mainfrom
feat/map-run

nvasiu commented Aug 20, 2026

Uh oh!

yaythomas left a comment

Uh oh!

yaythomas Aug 20, 2026

Uh oh!

yaythomas Aug 20, 2026

Uh oh!

yaythomas Aug 20, 2026

Uh oh!

yaythomas Aug 20, 2026

Uh oh!

yaythomas Aug 20, 2026

Uh oh!

yaythomas Aug 20, 2026

Uh oh!

yaythomas Aug 21, 2026

Uh oh!

yaythomas Aug 21, 2026

Uh oh!

yaythomas Aug 21, 2026

Uh oh!

yaythomas Aug 21, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

		@@ -0,0 +1,502 @@
		"""Implement the Durable map run operation."""

		@@ -0,0 +1,241 @@
		"""Authoring helpers for distributed map processor and reader Lambda functions.

		logger = logging.getLogger(__name__)


		@functools.lru_cache(maxsize=1)

Conversation

nvasiu commented Aug 20, 2026

Summary

Changes

Tests

TODO

Future Tasks

Uh oh!

yaythomas left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants